udsink

Documentation

Index

• type ApplyUDSinkErr
  • func (e ApplyUDSinkErr) Error() string
  • func (e ApplyUDSinkErr) IsInternalErr() bool
  • func (e ApplyUDSinkErr) IsUserUDSinkErr() bool
• type InternalErr
• type Option
  • func WithLogger(log *zap.SugaredLogger) Option
• type SinkApplier
• type UDSgRPCBasedUDSink
  • func NewUDSgRPCBasedUDSink(client sinkclient.Client) *UDSgRPCBasedUDSink
  • func (u *UDSgRPCBasedUDSink) ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error
  • func (u *UDSgRPCBasedUDSink) CloseConn(ctx context.Context) error
  • func (u *UDSgRPCBasedUDSink) IsHealthy(ctx context.Context) error
  • func (u *UDSgRPCBasedUDSink) WaitUntilReady(ctx context.Context) error
• type UserDefinedSink
  • func NewUserDefinedSink(vertexInstance *dfv1.VertexInstance, fromBuffer isb.BufferReader, ...) (*UserDefinedSink, error)
  • func (s *UserDefinedSink) Close() error
  • func (s *UserDefinedSink) ForceStop()
  • func (s *UserDefinedSink) GetName() string
  • func (s *UserDefinedSink) GetPartitionIdx() int32
  • func (s *UserDefinedSink) IsFull() bool
  • func (s *UserDefinedSink) Start() <-chan struct{}
  • func (s *UserDefinedSink) Stop()
  • func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]isb.Offset, []error)

Types

type ApplyUDSinkErr

type ApplyUDSinkErr struct {
	UserUDSinkErr bool
	Message       string
	InternalErr
}

ApplyUDSinkErr represents any UDSink related error

func (ApplyUDSinkErr) Error

func (e ApplyUDSinkErr) Error() string

func (ApplyUDSinkErr) IsInternalErr

func (e ApplyUDSinkErr) IsInternalErr() bool

IsInternalErr is true if this is a platform issue. This is a blocking error.

func (ApplyUDSinkErr) IsUserUDSinkErr

func (e ApplyUDSinkErr) IsUserUDSinkErr() bool

IsUserUDSinkErr is true if the problem is due to the user code in the UDSink.

type InternalErr

type InternalErr struct {
	Flag        bool
	MainCarDown bool
}

InternalErr represents errors internal to the platform

type Option

type Option func(*UserDefinedSink) error

func WithLogger

func WithLogger(log *zap.SugaredLogger) Option

type SinkApplier

type SinkApplier interface {
	ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error
}

SinkApplier applies the sink on the read message and gives back a response. Any UserError will be retried here, while InternalErr can be returned and could be retried by the callee.

type UDSgRPCBasedUDSink

type UDSgRPCBasedUDSink struct {
	// contains filtered or unexported fields
}

UDSgRPCBasedUDSink applies user defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink.

func NewUDSgRPCBasedUDSink

func NewUDSgRPCBasedUDSink(client sinkclient.Client) *UDSgRPCBasedUDSink

NewUDSgRPCBasedUDSink returns UDSgRPCBasedUDSink

func (*UDSgRPCBasedUDSink) ApplySink

func (u *UDSgRPCBasedUDSink) ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error

func (*UDSgRPCBasedUDSink) CloseConn

func (u *UDSgRPCBasedUDSink) CloseConn(ctx context.Context) error

CloseConn closes the gRPC client connection.

func (*UDSgRPCBasedUDSink) IsHealthy

func (u *UDSgRPCBasedUDSink) IsHealthy(ctx context.Context) error

IsHealthy checks if the udsink is healthy.

func (*UDSgRPCBasedUDSink) WaitUntilReady

func (u *UDSgRPCBasedUDSink) WaitUntilReady(ctx context.Context) error

WaitUntilReady waits until the udsink is connected.

type UserDefinedSink

type UserDefinedSink struct {
	// contains filtered or unexported fields
}

func NewUserDefinedSink

func NewUserDefinedSink(vertexInstance *dfv1.VertexInstance,
	fromBuffer isb.BufferReader,
	fetchWatermark fetch.Fetcher,
	publishWatermark publish.Publisher,
	idleManager wmb.IdleManager,
	udsink SinkApplier,
	opts ...Option) (*UserDefinedSink, error)

NewUserDefinedSink returns genericSink type.

func (*UserDefinedSink) Close

func (s *UserDefinedSink) Close() error

func (*UserDefinedSink) ForceStop

func (s *UserDefinedSink) ForceStop()

func (*UserDefinedSink) GetName

func (s *UserDefinedSink) GetName() string

func (*UserDefinedSink) GetPartitionIdx

func (s *UserDefinedSink) GetPartitionIdx() int32

GetPartitionIdx returns the partition index. for sink it is always 0.

func (*UserDefinedSink) IsFull

func (s *UserDefinedSink) IsFull() bool

func (*UserDefinedSink) Start

func (s *UserDefinedSink) Start() <-chan struct{}

func (*UserDefinedSink) Stop

func (s *UserDefinedSink) Stop()

func (*UserDefinedSink) Write

func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]isb.Offset, []error)

Write writes to the UDSink container.